package cn.lsh.kafka.consumer;

import cn.lsh.kafka.db.service.BusinessService;
import cn.lsh.kafka.db.service.OffsetService;
import cn.lsh.kafka.db.service.ServiceFactory;
import cn.lsh.kafka.db.service.impl.BusinessServiceImpl;
import cn.lsh.kafka.listener.MyConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.time.LocalTime;
import java.util.Collections;
import java.util.Properties;

public class OffsetConsumer {

	public static void main(String[] args) {
		String groupId = "group_1";
		Properties properties = new Properties();
		properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "node00:9092,node01:9092,node02:9092");
		properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.IntegerDeserializer");
		properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

		properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
		// properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
		properties.setProperty(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "5");
		properties.setProperty(ConsumerConfig.EXCLUDE_INTERNAL_TOPICS_CONFIG, "true");

		//关闭自动提交offset
		properties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

		KafkaConsumer<Integer, String> consumer = new KafkaConsumer<>(properties);

		//订阅topic时设置自定义的Rebalance监听器
		consumer.subscribe(Collections.singletonList("ddd-1"), new MyConsumerRebalanceListener(consumer));

		OffsetService offsetService = ServiceFactory.buildService(OffsetService.class);
		BusinessService busiService = new BusinessServiceImpl();
		while (true) {
			ConsumerRecords<Integer, String> records = consumer.poll(Duration.ofSeconds(1));
			if (records.count() > 0) {
				System.out.println("当前批次拉取数据：" + records.count() + "，时间：" + LocalTime.now());
			}
			for (ConsumerRecord<Integer, String> record : records) {
				System.out.println(record.offset() + "     " + record.key() + "     " + record.value());
				TopicPartition partition = new TopicPartition(record.topic(), record.partition());
				MyConsumerRebalanceListener.currentOffset.put(partition, record.offset());
				//业务处理
				busiService.deal(record.value());
			}
			//保存offset到mysql
			MyConsumerRebalanceListener.commitOffset(groupId, offsetService);
			//提交事务，让业务处理和提交offset保持一个事物
			ServiceFactory.commit();
			consumer.close();
		}
	}
}
